package beautiful.butterfly.drds.data_exchange.plugin_collector;


import beautiful.butterfly.drds.data_exchange.constant.TaskType;
import beautiful.butterfly.drds.data_exchange.data_exchange.record.column.Record;
import beautiful.butterfly.drds.data_exchange.error_code.DataExchangeException;
import beautiful.butterfly.drds.data_exchange.error_code.FrameworkErrorCode;
import beautiful.butterfly.drds.data_exchange.report.Message;
import beautiful.butterfly.drds.data_exchange.report.Messages;
import beautiful.butterfly.drds.data_exchange.util.Configuration;
import lombok.extern.slf4j.Slf4j;


@Slf4j
public class TaskPluginCollector extends AbstractTaskPluginCollector
{


    private Message message;

    private Configuration configuration;

    private TaskType taskType;

    public TaskPluginCollector(Configuration configuration, Message message,
                               TaskType taskType)
    {
        this.configuration = configuration;
        this.message = message;
        this.taskType = taskType;
    }

    public Message getMessage()
    {
        return message;
    }

    public Configuration getConfiguration()
    {
        return configuration;
    }


    public void collectDirtyRecord(Record dirtyRecord, Throwable throwable,
                                   String errorMessage)
    {

        if (null == dirtyRecord)
        {
            log.warn("脏数据record=null.");
            return;
        }

        if (this.taskType.equals(TaskType.reade))
        {
            this.message.increase(
                    Messages.read_failed_records, 1);
            this.message.increase(
                    Messages.read_failed_bytes, dirtyRecord.getByteSize());
        } else if (this.taskType.equals(TaskType.write))
        {
            this.message.increase(
                    Messages.write_failed_records, 1);
            this.message.increase(
                    Messages.write_failed_bytes, dirtyRecord.getByteSize());
        } else
        {
            throw DataExchangeException.asDataExchangeException(
                    FrameworkErrorCode.RUNTIME_ERROR,
                    String.format("不知道的插件类型[%s].", this.taskType));
        }
    }
}
